package com.atguigu.tingshu.album.service.impl;

import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackInfoMapper;
import com.atguigu.tingshu.album.service.AlbumAttributeValueService;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.*;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

	@Autowired
	private AlbumInfoMapper albumInfoMapper;

	@Autowired
	private AlbumStatMapper albumStatMapper;

	@Autowired
	private AlbumAttributeValueMapper albumAttributeValueMapper;

	@Autowired
	private AlbumAttributeValueService albumAttributeValueService;

	@Autowired
	private KafkaService kafkaService;

	@Autowired
	private TrackInfoMapper trackInfoMapper;

	@Autowired
	private RedissonClient redissonClient;

	@Autowired
	private RedisTemplate redisTemplate;

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void saveAlbumInfo(AlbumInfoVo albumInfoVo,Long userId) {
		//	album_info
		try {
			//	创建专辑对象
			AlbumInfo albumInfo = new AlbumInfo();
			//	进行属性拷贝
			BeanUtils.copyProperties(albumInfoVo, albumInfo);
			albumInfo.setUserId(userId);
			albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
			//	设置免费的集数.
			if (!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfoVo.getPayType())){
				//	只要收费就设置前5集为免费试听集数
				albumInfo.setTracksForFree(5);
			}
			//	当执行插入完成之后，获取到专辑Id
			//	在实体类id属性上 @TableId(type = IdType.AUTO) 表示插入完成之后能够自动获取到主键自增的Id.
			albumInfoMapper.insert(albumInfo);

			//  album_attribute_value
			List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
			//	判断
			if (!CollectionUtils.isEmpty(albumAttributeValueVoList)){
				/*
				//这个方法相当于Stream流
				//声明一个集合
				List<AlbumAttributeValue> attributeValueList = new ArrayList<>();
				//	循环遍历这个集合保存数据.
				for (AlbumAttributeValueVo albumAttributeValueVo : albumAttributeValueVoList) {
					//	创建一个专辑对应的属性与属性值对象
					AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
					albumAttributeValue.setAlbumId(albumInfo.getId());
					albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
					albumAttributeValue.setValueId(albumAttributeValue.getValueId());
					//albumAttributeValueMapper.insert(albumAttributeValue);
					attributeValueList.add(albumAttributeValue);
				}*/

				List<AlbumAttributeValue> attributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
					//创建对象
					AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
					//赋值
					albumAttributeValue.setAlbumId(albumInfo.getId());
					albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
					albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
					//返回对象
					return albumAttributeValue;
				}).collect(Collectors.toList());

				//批量保存：albumAttributeValueMapper album_attribute_value
				//利用mybatis-plus中  IService  ServiceIMPl 这个类处理 批量保存方法
				albumAttributeValueService.saveBatch(attributeValueList);
			}

			//	album_stat
			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_PLAY);
			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_SUBSCRIBE);
			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_BROWSE);
			this.saveAlbumStat(albumInfo.getId(), SystemConstant.ALBUM_STAT_COMMENT);

			//发送上架信息
			if("1".equals(albumInfoVo.getIsOpen())){
				//获取布隆过滤器
				RBloomFilter<Object> bloomFilter = this.redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
				//商品上架
				bloomFilter.add(albumInfo.getId());
				this.kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER,albumInfo.getId().toString());
			}
		} catch (BeansException e) {
			log.error("保存专辑失败");
			throw new RuntimeException(e);
		}

	}

	public void saveAlbumStat(Long albumId, String statPlay) {
		//	创建统计对象
		AlbumStat albumStat = new AlbumStat();
		albumStat.setAlbumId(albumId);
		albumStat.setStatType(statPlay);
		albumStat.setStatNum(0);
		albumStatMapper.insert(albumStat);
	}


	@Override
	public IPage<AlbumListVo> getUserAlbumPage(Page<AlbumListVo> albumListVoPage, AlbumInfoQuery albumInfoQuery) {
		//  根据业务分析：需要  album_info   album_stat

		return albumInfoMapper.selectUserAlbumPage(albumListVoPage, albumInfoQuery);
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public void removeAlbumInfo(Long albumId) {
		/*
		album_attribute_value
		album_info
		album_stat
		 */
		//删除分为逻辑删除 update album_info  set is_deleted = ? where id = ?
		// 用的是物理删除      delete from album_info where ablum_id = ?
		albumInfoMapper.deleteById(albumId);
		//where album_id = ?
		albumStatMapper.delete(new LambdaQueryWrapper<AlbumStat>().eq(AlbumStat::getAlbumId, albumId));
		//album_attribute_value  ablum_id
		albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
	}


	@Override
	public AlbumInfo getAlbumInfoById(Long albumId) {
		//	查询缓存：String set key value;
		String key = RedisConstant.ALBUM_INFO_PREFIX + albumId;
		//  从缓存中获取数据
		try {
			AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(key);
			//  判断当前专辑对象是否为空
			if (null == albumInfo) {
				//  缓存中没有数据，查询数据库，避免缓存击穿要加分布式锁！
				//  声明一个锁的key
				String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
				RLock lock = redissonClient.getLock(lockKey);
				//  如何lock加锁！
				lock.lock();
				//  执行业务逻辑。
				try {
					//  获取到锁的时候，最好再查询一次缓存。
					albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(key);
					log.info("查询了缓存....");
					if (null != albumInfo) {
						return albumInfo;
					}
					//  查询数据库
					albumInfo = getAlbumInfoDB(albumId);
					//  判断当前这个对象是否存在.
					if (null == albumInfo) {
						//  返回数据
						this.redisTemplate.opsForValue().set(key, new AlbumInfo(), RedisConstant.ALBUM_TEMPORARY_TIMEOUT, TimeUnit.SECONDS);
						//  返回空对象
						return new AlbumInfo();
					} else {
						//  数据库中有数据
						this.redisTemplate.opsForValue().set(key, albumInfo, RedisConstant.ALBUM_TIMEOUT, TimeUnit.SECONDS);
						return albumInfo;
					}
				} finally {
					//  解锁
					lock.unlock();
				}
			} else {
				return albumInfo;
			}
		} catch (Exception e) {
			log.error("出现了异常{}专辑Id：\t", albumId);
			//  调用sendMsg(phoneNumber);
			e.printStackTrace();
		}
		//	查询数据库
		return getAlbumInfoDB(albumId);
	}

	//	查询数据库.
	@Nullable
	private AlbumInfo getAlbumInfoDB(Long albumId) {
		//	先获取到专辑信息表数据 album_info
		//  select * from album_info where id = ? and is_deleted = 0;
		AlbumInfo albumInfo = albumInfoMapper.selectById(albumId);
		//	获取到 专辑属性与属性值数据集合.
		if (null != albumInfo) {
			//	想办法获取到属性与属性值集合数据.
			//	根据 album_id 查询 album_attribute_value 集合.
			//  select * from album_attribute_value where album_id = ?;
			List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
			albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
		}
		//	返回对象数据
		return albumInfo;
	}

	/**
	 * 保证redis和mysql一致性 采用延迟双删策略 1.先删除缓存,2.修改db数据,3.睡眠100毫秒,4.删除缓存
	 * @param albumId
	 * @param albumInfoVo
	 */
	@Override
	@Transactional(rollbackFor = Exception.class)
	public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {
		//先删除缓存的key
		String albumInfoKey = RedisConstant.ALBUM_INFO_PREFIX+"["+albumId+"]";
		this.redisTemplate.delete(albumInfoKey);
		//修改简介   相当于改album_info
		AlbumInfo albumInfo = new AlbumInfo();
		//给albumInfo 对象赋值
		BeanUtils.copyProperties(albumInfoVo, albumInfo);
		albumInfo.setId(albumId);
		albumInfoMapper.updateById(albumInfo);

		//修改标签   相当于改album_attribute_value
		//先删除   后新增
		albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));

		//  album_attribute_value
		List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
		//	判断
		if (!CollectionUtils.isEmpty(albumAttributeValueVoList)){
			List<AlbumAttributeValue> attributeValueList = albumAttributeValueVoList.stream().map(albumAttributeValueVo -> {
				//创建对像
				AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
				//赋值
				albumAttributeValue.setAlbumId(albumInfo.getId());
				albumAttributeValue.setAttributeId(albumAttributeValueVo.getAttributeId());
				albumAttributeValue.setValueId(albumAttributeValueVo.getValueId());
				//返回对象
				return albumAttributeValue;
			}).collect(Collectors.toList());
			//批量保存：albumAttributeValueMapper album_attribute_value
			//利用mybatis-plus中  IService  ServiceIMPl 这个类处理 批量保存方法
			albumAttributeValueService.saveBatch(attributeValueList);
		}
		try {
			Thread.sleep(100);
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		}
		this.redisTemplate.delete(albumInfoKey);
		//	判断是否上架：
		if ("1".equals(albumInfoVo.getIsOpen())){
			//获取布隆过滤器
			RBloomFilter<Object> bloomFilter = this.redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
			//上架专辑
			bloomFilter.add(albumInfo.getId());
			//	上架
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
		} else {
			//	下架
			kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, albumId.toString());
		}
	}

	@Override
	@Transactional(rollbackFor = Exception.class)
	public List<AlbumInfo> findUserAllAlbumList(Long userId) {
		//根据用户ID 查询专辑列表
		//List<AlbumInfo> albumInfoList = albumInfoMapper.selectList(new LambdaQueryWrapper<AlbumInfo>().eq(AlbumInfo::getUserId, userId));
		/*
		//第一种方案
		Page<AlbumInfo> albumInfoPage = new Page<>(1,20);
		LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<AlbumInfo>()
				.eq(AlbumInfo::getUserId, userId)
				.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle)
				.orderByDesc(AlbumInfo::getCreateTime);//根据用户id  降序排序
		Page<AlbumInfo> selectPage = albumInfoMapper.selectPage(albumInfoPage, queryWrapper);
		return selectPage.getRecords();*/

		//第二种方案
		LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<AlbumInfo>()
				.eq(AlbumInfo::getUserId, userId)
				.select(AlbumInfo::getId,AlbumInfo::getAlbumTitle)
				.orderByDesc(AlbumInfo::getId)
				.last("limit 20");//根据用户id  降序排序
		//返回数据
		return albumInfoMapper.selectList(queryWrapper);
	}

	@Override
	public List<AlbumAttributeValue> findAlbumAttributeValue(Long albumId) {
		//根据·专辑ID 查询 专辑属性值 的数据
		//创建一个条件对象
		/*LambdaQueryWrapper<AlbumAttributeValue> lambdaQueryWrapper = new LambdaQueryWrapper<>();
		//通过获取到AlbumAttributeValue中的getAlbumId   进而返回albumId
		lambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId,albumId);
		List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(lambdaQueryWrapper);*/

		//返回数据
		return albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId,albumId));
	}

	@Override
	public AlbumStatVo getAlbumStatVo(Long albumId) {
		//	调用mapper
		return albumInfoMapper.selectAlbumStatVo(albumId);
	}

	/**
	 * 更新声音播放量
	 *
	 * @param trackStatMqVo
	 */
	@Override
	@Transactional(rollbackFor = Exception.class)
	public void trackStatUpdate(TrackStatMqVo trackStatMqVo) throws  Exception{
		//	更新数据声音： update track_stat set stat_num = stat_num + 1 where track_id = 1 and stat_type = '0701';
		trackInfoMapper.trackStatUpdate(trackStatMqVo.getTrackId(), trackStatMqVo.getStatType(), trackStatMqVo.getCount());
		//	更新专辑：
		if (SystemConstant.TRACK_STAT_PLAY.equals(trackStatMqVo.getStatType())) {
			//	调用更新专辑播放量方法.
			albumInfoMapper.updateAlbumStat(trackStatMqVo.getAlbumId(), SystemConstant.ALBUM_STAT_PLAY, trackStatMqVo.getCount());
		}
	}
}
